package A_

import U.Utils
import rx.Observable
import rx.Subscriber
import rx.functions.Func1
import rx.internal.producers.SingleDelayedProducer

class all{
    /**
     * 使用kt的扩展函数，在Observable类上扩展some操作
     */
    private fun <T> Observable<T>.some(predicate: Func1<T, Boolean>): Observable<Int> {
        return this.lift(OperatorSome<T>(predicate))
    }


    fun main(args: Array<String>) {
        //        amb();
        all()
    }

    /**
     * 是否所有的item都符合条件
     */
    private fun all() {
        val observable = Observable.from(Utils.INTS)
        observable.all { integer -> integer < 2 }.subscribe { }


        //自定义一个 some operator,统计符合条件的item数量(rx 里有类似的统计操作符？)
        Observable.from(arrayOf(1, 2, 101, 200, 7, 130))
                .some(Func1 { o -> o > 100 })
                .map { i -> i }
                .subscribe({
                    println("大于一百的 数字有 " + it + "个")

                })


    }

    open class OperatorSome<T>(internal val predicate: Func1<in T, Boolean>) : Observable.Operator<Int, T> {

        private var some: Int = 0

        override fun call(child: Subscriber<in Int>): Subscriber<in T> {
            val producer = SingleDelayedProducer(child)
            val s = object : Subscriber<T>() {
                override fun onCompleted() {
                    producer.setValue(some)//内部会调用onComplete
                }

                override fun onError(e: Throwable) {
                    child.onError(e)
                }

                override fun onNext(t: T) {
                    if (predicate.call(t))
                        some++
                }
            }
            child.add(s)
            child.setProducer(producer)
            return s
        }
    }
}
